package com.zdxlz.fcmp.common.mqtt.adapter;

import cn.hutool.core.util.IdUtil;
import com.zdxlz.fcmp.common.core.exception.BizException;
import com.zdxlz.fcmp.common.core.util.JsonUtil;
import com.zdxlz.fcmp.common.mqtt.config.MqttConfig;
import com.zdxlz.fcmp.common.mqtt.dto.IctMqttRequest;
import com.zdxlz.fcmp.common.mqtt.dto.IctMqttResponse;
import com.zdxlz.fcmp.common.mqtt.handler.CustomMqttCallbackHandler;
import com.zdxlz.fcmp.common.mqtt.handler.MqttMessageRouter;
import jakarta.annotation.PostConstruct;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.mqttv5.client.MqttAsyncClient;
import org.eclipse.paho.mqttv5.client.MqttConnectionOptions;
import org.eclipse.paho.mqttv5.client.persist.MemoryPersistence;
import org.eclipse.paho.mqttv5.common.MqttException;
import org.eclipse.paho.mqttv5.common.MqttMessage;
import org.eclipse.paho.mqttv5.common.packet.MqttProperties;
import org.springframework.core.task.TaskExecutor;
import org.springframework.stereotype.Component;

import java.util.concurrent.*;
import java.util.function.BiConsumer;

import static com.baomidou.mybatisplus.core.toolkit.StringPool.HASH;
import static com.baomidou.mybatisplus.core.toolkit.StringPool.UNDERSCORE;
import static com.zdxlz.fcmp.common.core.constant.MqttMsgConstants.*;

/**
 * @author: aman
 * @date 2025/2/10 17:17
 * @description
 */

@Slf4j
@Component
public class Mqttv5Adapter implements ProtocolAdapter {

    @Resource
    private MqttConfig mqttConfig;

    @Resource
    private MqttMessageRouter router;

    @Resource
    private TaskExecutor mqttMsgTaskExecutor;

    private MqttAsyncClient client;

    private final ConcurrentMap<String, CompletableFuture<IctMqttResponse<?>>> pendingRequests = new ConcurrentHashMap<>();


    @Override
    public void initialize() {
        if (client != null) {
            return;
        }
        try {
            client = new MqttAsyncClient(
                    mqttConfig.getBrokerUrl(),
                    mqttConfig.getClientIdPrefix() + UNDERSCORE + IdUtil.simpleUUID(),
                    new MemoryPersistence()
            );

            MqttConnectionOptions options = new MqttConnectionOptions();
            options.setAutomaticReconnect(true);
            options.setCleanStart(false);
            options.setUserName(mqttConfig.getUsername());
            options.setPassword(mqttConfig.getPassword().getBytes());

            CustomMqttCallbackHandler handler = new CustomMqttCallbackHandler(mqttMsgTaskExecutor, pendingRequests);
            client.setCallback(handler);
            client.connect(options).waitForCompletion();
            log.info("【MQTT】客户端初始化成功");
            String reportTopic = MQTT_QUANTUM_REPORT_TOPIC + HASH;
            String replyTopic = MQTT_QUANTUM_REPLY_TOPIC + HASH;

            // 订阅的主题数组
            String[] topics = {reportTopic, replyTopic};
            int[] qosLevels = {1, 1};

            client.subscribe(topics, qosLevels);

            log.info("【MQTT】订阅Topic：【{}】,QoS: {}", topics, qosLevels);
        } catch (MqttException e) {
            log.error("初始化MQTT客户端失败", e);
            throw new BizException("初始化MQTT客户端失败！");
        }

    }


    @Override
    public void publichTopic(String deviceMac, IctMqttRequest<?> config) {
        String topic = MQTT_QUANTUM_PUBLISH_TOPIC + deviceMac;
        String messageId = config.getId();
        String json = JsonUtil.toJson(config);
        MqttMessage message = new MqttMessage(json.getBytes());
        message.setQos(1);
        message.setRetained(false);

        try {
            client.publish(topic, message).waitForCompletion();
        } catch (MqttException e) {
            log.error("发布消息失败：{}", topic);
            throw new RuntimeException(e);
        }
        log.info("发布消息到设备, topic: {}, msgId: {}", topic, messageId);
        log.debug("发送消息数据: {}", json);
    }

    @Override
    public CompletableFuture<IctMqttResponse<?>> applyConfigurationAsync(String deviceMac, IctMqttRequest<?> config) {
        String topic = MQTT_QUANTUM_PUBLISH_TOPIC + deviceMac;
        String responseTopic = MQTT_QUANTUM_REPLY_TOPIC + deviceMac;
        CompletableFuture<IctMqttResponse<?>> future = new CompletableFuture<>();
        pendingRequests.put(responseTopic, future);
        String messageId = config.getId();
        String json = JsonUtil.toJson(config);
        MqttMessage message = new MqttMessage(json.getBytes());
        message.setQos(1);
        message.setRetained(false);
        message.setProperties(new MqttProperties());

        message.getProperties().setResponseTopic(responseTopic);
        // 设置关联数据
        message.getProperties().setCorrelationData(messageId.getBytes());
        try {
            client.publish(topic, message).waitForCompletion();
            log.info("发布消息到设备, topic: {}, msgId: {}", topic, messageId);
            log.debug("发送消息数据: {}", json);
            log.info("订阅响应topic: {}", responseTopic);
            // 超时机制，避免设备不响应导致的死锁
            CompletableFuture.delayedExecutor(MQTT_REPLY_TIMEOUT, TimeUnit.SECONDS).execute(() -> {
                if (!future.isDone()) {
                    pendingRequests.remove(responseTopic);
                    future.completeExceptionally(new TimeoutException(MQTT_REPLY_TIMEOUT_MSG));
                }
            });
        } catch (MqttException e) {
            future.completeExceptionally(e);
        }

        return future;
    }


    @Override
    public void subscribeState(String topic, BiConsumer<String, MqttMessage> handler) throws MqttException {
        client.subscribe(topic, 1);
    }

    @Override
    public boolean isHealthy() {
        return false;
    }

    @Override
    public void disconnect() {

    }

    @Override
    public String getProtocolType() {
        return null;
    }

    @PostConstruct
    public void init() {
        initialize();
    }

}
